package com.topn;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.util.Arrays;

public class MyTopN {

    public static void main(String[] args) throws Exception {

        // 解析项目中的配置文件
        Configuration conf = new Configuration(true);

        //工具类帮我们把-D 等等的属性直接set到conf，会留下commandOptions
        GenericOptionsParser parser = new GenericOptionsParser(conf, args);
        String[] othargs = parser.getRemainingArgs();

        System.out.println(Arrays.toString(othargs));

        //让框架知道是windows异构平台运行
        conf.set("mapreduce.app-submission.cross-platform", "true");
        // 配置文件中的 mapreduce.framework.name 属性默认是 yarn，也就是集群模式
        conf.set("mapreduce.framework.name", "local");

        Job job = Job.getInstance(conf);

        // /data/mapjoin/tianqi /data/mapjoin/location /data/mapjoin/output
        // job.setJar("D:\\bigdata_project\\bigdata_test\\target\\bigdata_test-1.0-SNAPSHOT.jar");
        //必须必须写的,反推这个类属于哪个jar包
        job.setJarByClass(MyTopN.class);

        job.setJobName("topN");

        Path infile = new Path(othargs[0]);
        // 可以添加多个输入路径,TextInputFormat表示读取文本信息的格式
        TextInputFormat.addInputPath(job, infile);

        Path outfile = new Path(othargs[2]);
        // 如果路径存在，先删除path file
        if (outfile.getFileSystem(conf).exists(outfile)) {
            outfile.getFileSystem(conf).delete(outfile, true);
        }
        // 只能设置一个输出路径
        TextOutputFormat.setOutputPath(job, outfile);


        job.setMapperClass(TMapper.class);
        // 设置map输出的key，value类型
        job.setMapOutputKeyClass(Tkey.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 排序比较器
        job.setSortComparatorClass(TSortComparator.class);

        // 设置分区器
        job.setPartitionerClass(TPartitioner.class);

        // 分组比较器
        job.setGroupingComparatorClass(TGroupComparator.class);


        job.setReducerClass(TReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 小表所在的目录
        job.addCacheFile(new Path(othargs[1]).toUri());

        // 设置分区数
        // job.setNumReduceTasks(3);

        job.waitForCompletion(true);

    }

}
